#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <unistd.h>
#include <fcntl.h>

#include <boost/bind.hpp>

#include "Loop.h"
#include "Server.h"
#include "Task.h"
#include "Log.h"

namespace sevent{

Loop::Loop(const string ip, const int port)
:_pollThread(boost::bind(&Loop::pollThreadInit, this), boost::bind(&Loop::pollThreadRun, this), boost::bind(&Loop::pollThreadStop, this)), _started(false), _listenAddr(ip, port){
	pthread_mutex_init(&_mutex, NULL);
	pthread_cond_init(&_cond, NULL);
}

Loop::~Loop(){
	if(_started){
		_started = false;
		_pollThread.stop();
	}
	tcpClose(_listenSock);
	tcpClose(_epfd);
	pthread_mutex_destroy(&_mutex);
	pthread_cond_destroy(&_cond);
}


int Loop::init(){
	_epfd = epoll_create(1024);
	LOG_DEBUG_VA("epoll fd create epfd: %d", _epfd);
	if (-1 == _epfd) {
		LOG_ERROR_VA("epoll fd create error.");
		return -1;
	}
	int ret = 0;
	_listenSock = socket(AF_INET, SOCK_STREAM, 0);
	LOG_DEBUG_VA("listen socket: %d", _listenSock);
	if (_listenSock < 0) {
		LOG_ERROR_VA("listening socket open error.");
		return -1;
	}
	int optval = 1;
	ret = setsockopt(_listenSock, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(int));
	if (ret < 0) {
		LOG_ERROR_VA("set socket option error.");
		return -1;
	}
	ret = ::bind(_listenSock, (struct sockaddr *)(&_listenAddr.addr()), _listenAddr.length());
	if (ret < 0) {
		LOG_ERROR_VA("bind error.");
		sevent::tcpClose(_listenSock);
		return -1;
	}
	int sockFlag = fcntl(_listenSock, F_GETFL, 0);
	fcntl(_listenSock, F_SETFL, sockFlag | O_NONBLOCK);

	ret = listen(_listenSock, 128);
	if (ret < 0) {
		char errbuf[1024];
		strerror_r(errno, errbuf, 1023);
		LOG_ERROR_VA("listening socket error: %s", errbuf);
		sevent::tcpClose(_listenSock);
		return -1;
	}
	_eplistenevent.data.u64 = 0UL;
	_eplistenevent.data.fd = _listenSock;
	_eplistenevent.events = EPOLLIN | EPOLLET;
	ret = epoll_ctl(_epfd, EPOLL_CTL_ADD, _listenSock, &_eplistenevent);
	if (ret < 0) {
		LOG_ERROR_VA("epoll add listen sock error.");
		sevent::tcpClose(_listenSock);
		return -1;
	}
	
	return ret;
}

int Loop::pollThreadInit(){
	LOG_DEBUG_VA("Loop pollThreadInit");
	return 0;
}
int Loop::pollThreadStop(){
	LOG_DEBUG_VA("Loop pollThreadStop");
	return 0;
}
int Loop::pollThreadRun(){
	LOG_DEBUG_VA("Loop pollThreadRun");
	int ret = 0;
	while (_started) {
		ret = checkTask();
		if (ret < 0) {
			LOG_ERROR_VA("check task error");
		}
	}
	LOG_INFO_VA("Loop stop");
	return 0;
}

int Loop::checkTask(){
	int ret = 0;
	int numfd = sevent::tcpEpollWait(_epfd, _epevents, 1024, -1);
	if (numfd < 0) {
		return numfd;
	}
	//dispatch ready fd
	struct sockaddr_in peerAddr;
	socklen_t peerAddrLen = sizeof(peerAddr);
	LOG_DEBUG_VA("Loop epoll_wait num: %d", numfd);
	for (int i = 0; i < numfd; ++i) {
		if (_epevents[i].events & EPOLLIN) {
			if(_epevents[i].data.fd == _listenSock) {
				std::vector<Connect *> connectVec;
				int ret = sevent::tcpAccept(_listenSock, connectVec);
				if (ret < 0) {
					LOG_ERROR_VA("one task fd failed to accept.");
					continue;
				}
				for(std::vector<Connect *>::iterator iter = connectVec.begin(); iter != connectVec.end(); ++iter) {
					ret = prepareConnect(*iter);
					if (-1 == ret) {
						LOG_ERROR_VA("Loop refuse to accept fd: %d ", (*iter)->fd());
						//sevent::tcpClose((*iter)->fd()); //do this in delete Connect
						delete (*iter);
						continue;
					}
					int sockFlag = fcntl((*iter)->fd(), F_GETFL, 0);
					fcntl((*iter)->fd(), F_SETFL, sockFlag | O_NONBLOCK);
					ret = sevent::tcpEpollAdd(_epfd, (*iter)->fd(), NULL, EPOLLIN | EPOLLET);
					if (-1 == ret) {
						LOG_ERROR_VA("Loop %s add epoll failed fd: %d", name().c_str(), (*iter)->fd());
						//sevent::tcpClose((*iter)->fd()); //do this in delete Connect
						delete (*iter);
						continue;
					}else {
						LOG_DEBUG_VA("Loop %s accepted Connect %x:%d:%d <%s:%d>", 
								name().c_str(), (*iter), (*iter)->fd(), (*iter)->count(), (*iter)->peerAddr().ip().c_str(), (*iter)->peerAddr().port());
					}
				}
			}else{
				int fd = _epevents[i].data.fd;
				LOG_DEBUG_VA("Loop %s can receive from %d", name().c_str(), fd);
				Connect *connect = getConnect(fd);
				Task *t = NULL;
				if(connect == NULL){
					LOG_ERROR_VA("fd: %d get Connect failed.", fd);
					deleteConnect(fd);
				}else{
					t = new Task(connect, NULL, boost::bind(&Loop::recvMessageTask, this, _1, _2)); //this Task: just recv from connect
					LOG_DEBUG_VA("Loop %s create RecvTask %x with Connect %x:%d:%d ", name().c_str(), t, connect, connect->fd(), connect->count());

					int ret = getIOThreadPool().addTask(t);
					if(ret != 0){
						LOG_ERROR_VA("Loop %s Add Task %x error %x:%d ", name().c_str(), t, connect, connect->fd());
						delete t;
					}
				}
			}
		}else if(_epevents[i].events & EPOLLHUP) {
			LOG_ERROR_VA("fd: %d get EPOLLHUP.", _epevents[i].data.fd);
			deleteConnect(_epevents[i].data.fd);
		} else if (_epevents[i].events & EPOLLERR) {
			LOG_ERROR_VA("fd: %d get EPOLLERR.", _epevents[i].data.fd);
			deleteConnect(_epevents[i].data.fd);
		} else if (_epevents[i].events & EPOLLOUT) {
			LOG_ERROR_VA("fd: %d get EPOLLOUT.", _epevents[i].data.fd);
		} else {
			LOG_ERROR_VA("fd: %d get something wrong.", _epevents[i].data.fd);
			deleteConnect(_epevents[i].data.fd);
		}
	}
	return ret;
}

int Loop::start(){
	LOG_INFO_VA("Loop starting...");
	int ret;
	
	ret = init();
	if(ret != 0)
		return -1;
	_started = true;
	_pollThread.start();
	_IOThreadPool.start(1);
	return 0;
}

int Loop::prepareConnect(Connect *connect){
	int fd = connect->fd();
	std::map<int, Connect *>::iterator iter;
	iter = _connectMap.find(fd);
	if(iter != _connectMap.end()){
		if(!iter->second->inUsing()){
			LOG_INFO_VA("Loop %s Connect %x:%d:%d ", name().c_str(), connect, fd, connect->count());
			iter->second->addCount();
			iter->second->setFd(fd);
			iter->second->setPeerAddr(connect->peerAddr());
			struct timeval t;
			gettimeofday(&t, NULL);
			iter->second->setConnectTime(t);
			iter->second->setLastActiveTime(t);
			iter->second->setLoop(this);
			return 0;
		}else{
			LOG_ERROR_VA("Loop %s Connect %x:%d in use: %d", name().c_str(), connect, fd, iter->second->count());
			return -1;
		}
	}else{
		connect->addCount();
		struct timeval t;
		gettimeofday(&t, NULL);
		connect->setConnectTime(t);
		connect->setLastActiveTime(t);
		connect->setLoop(this);
		std::pair<map<int, Connect *>::iterator, bool> insertPair;
		//lock(); //just the listenThread will do this
		insertPair = _connectMap.insert(std::map<int, Connect *>::value_type(fd, connect));
		//unLock();
		if(insertPair.second != true)
			return -1;
		LOG_DEBUG_VA("Loop %s add Connect %x:%d:%d ", name().c_str(), connect, fd, connect->count());
	}
	return 0;
}

Connect *Loop::getConnect(int sockfd){
	std::map<int, Connect *>::iterator iter;
	iter = _connectMap.find(sockfd);
	if(iter == _connectMap.end()){
		//socket fd not found
		return NULL;
	}else{
		Connect *connect = iter->second;
		LOG_DEBUG_VA("Loop %s get Connect %x with fd %d ", name().c_str(), connect, connect->fd());
		return connect;
	}
}

int Loop::deleteConnect(int sockfd){
	std::map<int, Connect *>::iterator iter;
	iter = _connectMap.find(sockfd);
	if(iter == _connectMap.end()){
		//Connect fd not found, can't close
		return -1;
	}
	lock();
	_connectMap.erase(iter);
	unLock();
	return 0;
}

int Loop::recvMessageTask(Connect *connect, char *ptr){
	LOG_DEBUG_VA("recvMessageTask start execute");
	int len = 0;
	char *buf = new char[IOBUFSIZE]; //why new: keep the content till message task over
	len = connect->read(buf, IOBUFSIZE, 1000);
	if(len == 0){
		LOG_DEBUG_VA("Connect %x recv 0 byte, close the Connect", connect);
		connect->close();
		return 0;
	}

	buf[1023] = '\0';
//	msg->setHead(buffer.toHead());  //auto to do this in Buffer
	//len = connect->read(buffer, head.bodyLength(), 1000);
//	msg->setBody(buffer.toBody());  //auto to do this in Buffer
	/*
	if(SignalMessage)
		AppendToSignalThread()
	else
		AppendToServerThreadPool()
	*/
	Task *t = new Task(connect, buf, boost::bind(getServer()->getTaskFunc(), _1, _2));//the Task need a message
	LOG_DEBUG_VA("Loop %s create MessageTask %x with Connect %x:%d:%d ", name().c_str(), t, connect, connect->fd(), connect->count());
	int ret = getTaskThreadPool().addTask(t);
	if(ret != 0){
		LOG_ERROR_VA("Loop %s Add Task %x error %x:%d ", name().c_str(), t, connect, connect->fd());
		delete t;
		return -1;
	}
	return 0;
}

int Loop::sendMessageTask(Connect *connect, char *ptr){
	LOG_DEBUG_VA("sendMessageTask start execute");
	int len = 0;
	char *buf = new char[IOBUFSIZE]; //why new: keep the content till message task over
	memcpy(buf, ptr, IOBUFSIZE);
	len = connect->read(buf, IOBUFSIZE, 1000);
	if(len == 0){
		LOG_DEBUG_VA("Connect %x recv 0 byte, close the Connect", connect);
		return 0;
	}
	buf[1023] = '\0';
	//Buffer::toHead();  //auto to do this in Buffer
	//Buffer::toBody();  
	//if(SignalMessage)
	//	AppendToSignalThread()
	//else
	//	AppendToServerTaskThreadPool()
	
	Task *t = new Task(connect, buf, boost::bind(getServer()->getTaskFunc(), _1, _2));  //release the buf in Task::~Task
	LOG_DEBUG_VA("create Task %x with Connect %x", t, connect);
	int ret = getTaskThreadPool().addTask(t);
	if(ret != 0){
		LOG_ERROR_VA("Loop %s Add Task %x error %x:%d ", name().c_str(), t, connect, connect->fd());
		delete t;
		return -1;
	}
	return 0;
}

int Loop::lock(){
	pthread_mutex_lock(&_mutex);
	return 0;
}

int Loop::unLock(){
	pthread_mutex_unlock(&_mutex);
	return 0;
}

int Loop::wait(){
	_pollThread.wait();
	return 0;
}

int Loop::stop(){
	_started = false;
	_pollThread.wait();
	return 0;
}

std::string& Loop::name(){
	return getServer()->name();
}

ThreadPool& Loop::getTaskThreadPool(){
	return getServer()->getThreadPool();
}

ThreadPool& Loop::getIOThreadPool(){
	return _IOThreadPool;
}

}

